package com.sgcc.dlsc.kafka.producer.controller;

import com.sgcc.dlsc.kafka.producer.util.ResultUtil;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;

/**
 * @Author: 李孟帅
 * @CreateTime: 2019-12-31 08:48
 * @Description: ${Description}
 */
@RestController
public class ProducerController {
    private static final Logger LOGGER = LogManager.getLogger(ProducerController.class);
    @Resource
    private KafkaTemplate kafkaTemplate;

    @Resource
    private KafkaProducer kafkaProducer;

    @Value("${spring.kafka.topics}")
    private String topics;
    private int count = 1;

    @RequestMapping("/syncListLog")
    public ResultUtil sendList(@RequestBody Map<String, List<String>> datas) {
        try {
            kafkaProducer.beginTransaction();
            for (Map.Entry<String, List<String>> data : datas.entrySet()) {
                String topic = data.getKey();
                if (topic==null || "".equals(topic)){
                    return ResultUtil.result(500,"参数错误,主题不能为空");
                }
                LOGGER.info("发送批量消息到" + topic + "主题：" + count++);
                List<String> messages = data.getValue();
                for (String message : messages) {
                    ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);
                    kafkaProducer.send(record);
                }
            }
            kafkaProducer.commitTransaction();
        } catch (Exception e){
            LOGGER.error(e);
            LOGGER.error("数据同步失败，准备回滚...");
            kafkaProducer.abortTransaction();
            return ResultUtil.result(500,"事务提交失败，数据同步失败");
        }
        return ResultUtil.result(200, "事务提交成功，数据同步成功");
    }

    @RequestMapping("/syncLog")
    public ResultUtil syncLog(@RequestBody Map<String,String> data) {

        try {
            String topic = data.get("topic");
            if (topic==null || "".equals(topic)){
                return ResultUtil.result(500,"参数错误,主题不能为空");
            }
            String message = data.get("message");
            LOGGER.info("发送消息到" + topic + "主题：" + count++);
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);
            kafkaTemplate.send(record);
        } catch (Exception e) {
            LOGGER.error(e);
            LOGGER.error("数据同步失败");
            return ResultUtil.result(500,"数据同步失败");
        }


        return ResultUtil.result(200, "数据同步成功");
    }
}

